package zhao.exmaple;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;
import java.time.ZoneId;
import java.util.concurrent.TimeUnit;

import static java.time.Duration.ofSeconds;

public class FileSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        // 没两秒钟检测并关闭流
        env.enableCheckpointing(1000);
        // 参数为接口  意味着需要传入接口 实现类，ctrl+H查找实现类也有可能是接口内部的静态方法new一个实现类

        TextLineFormat textLineFormat = new TextLineFormat();
        Path path = new Path("E:\\note\\flink-project\\temp\\word.txt");
        FileSource<String> source = FileSource.forRecordStreamFormat(textLineFormat, path).build();

        DataStreamSource<String> dataSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "filesourcdemo");

        dataSource.print();

        // FileSink fileSink = FileSink
        //         .forRowFormat(new Path("E:\\note\\flink-project\\temp"), new SimpleStringEncoder<String>("UTF-8"))
        //         .withOutputFileConfig(
        //                 OutputFileConfig.builder().withPartPrefix("word-sink").withPartSuffix(".txt").build()
        //         )
        //         .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd HH", ZoneId.systemDefault()))
        //         // 自定义滚动策略 一小时时间桶
        //         .withRollingPolicy(
        //                 DefaultRollingPolicy.builder()
        //                         .withRolloverInterval(TimeUnit.SECONDS.toSeconds(10))  // 它至少包含 15 分钟的数据
        //                         // .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) // 最近 5 分钟没有收到新的记录
        //                         .withMaxPartSize(1024 * 1024) // 文件大小达到 1GB （写入最后一条记录后）
        //                         .build())
        //         .build();

        final FileSink<String> fileSink = FileSink
                .forRowFormat(new Path("E:\\note\\flink-project\\sinktemp"), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .build())
                .build();

        dataSource.sinkTo(fileSink);

        env.execute();

    }
}
